#include <cserver/session/upstream.h>
#include <cserver/context.h>
#include <boost/log/trivial.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/asio.hpp>

namespace csrv{
	namespace {
		const int K_retry_sleep_time = 10;
		const std::array<unsigned char, 15> K_CSHeart{
			0xb0, 0xb0, 0xb0,0xb0,	//head
				0xe0,				// command
				0x07,0x00,			//size
				'C','S','H','e','a','r','t',	// message
				0xf0				// end
		};
		const std::string K_startid_str="\xb0\xb0\xb0\xb0";
#pragma pack(push, 1)
		struct Header_t{
			std::uint8_t command;
			std::uint16_t length;	// little endian
		};
		union Header{
			Header_t head;
			std::uint8_t data[sizeof(Header_t)];
		};
		const size_t K_header_size = sizeof(Header_t);
#pragma pack(push)
	}

	using namespace boost::asio::ip;
	LORAUpstream::LORAUpstream(Context& ctx)
		: context(ctx)
		  , hb_deadline(ctx.ios_aserver)
	{
	}
	void LORAUpstream::start(){
		BOOST_LOG_TRIVIAL(info) << "Lora upstream started";
		boost::asio::spawn(context.ios_aserver, [&](boost::asio::yield_context yield) { run_(yield);});
	}
	void LORAUpstream::run_(boost::asio::yield_context yield){
		boost::system::error_code ec;

		boost::asio::steady_timer read_deadline(context.ios_aserver);

		std::shared_ptr<tcp::socket> socket;
		for(;;){
			socket = std::make_shared<tcp::socket>(context.ios_aserver);

			auto check_deadline = [this, socket, &read_deadline](const boost::system::error_code& tmec){
				if (!tmec && read_deadline.expires_at() <= std::chrono::steady_clock::now()){
					BOOST_LOG_TRIVIAL(warning) <<"Timeout, killed socket.";
					socket->close();
				}
			};
			read_deadline.expires_from_now(std::chrono::seconds(context.config.connect_timeout));
			read_deadline.async_wait(check_deadline);

			tcp::resolver resolver(context.ios_aserver);
			auto itr = resolver.async_resolve(tcp::resolver::query(context.config.aserver_site, boost::lexical_cast<std::string>(context.config.aserver_port)), yield[ec]);
			if (ec){
				BOOST_LOG_TRIVIAL(fatal) <<"Failed to resolve CServer site " << context.config.aserver_site;
				sleep_for_(K_retry_sleep_time * 6, yield);
				continue;
			}
			for (tcp::resolver::iterator end; itr != end; ++itr) {
				boost::asio::async_connect(*socket, itr, yield[ec]);
				if (!ec) break;
			}

			if (ec || !socket->is_open()){
				BOOST_LOG_TRIVIAL(warning) <<"Failed to connect to CServer site " << context.config.aserver_site << ":" << context.config.aserver_port << ". Message:" << ec.message();
				sleep_for_(K_retry_sleep_time, yield);
				continue;
			}
			BOOST_LOG_TRIVIAL(info) <<"Connected to CServer site " << context.config.aserver_site << ":" << context.config.aserver_port;

			boost::asio::spawn(context.ios_aserver, [socket,this](boost::asio::yield_context yield) { run_send_(yield, socket);});

			std::vector<char> v(4096);
			char* first = &*v.begin();
			char* pos = first;
			char* end = first + v.size();

			for(;;){
				read_deadline.expires_from_now(std::chrono::seconds(context.config.read_timeout));

				auto n = socket->async_read_some(boost::asio::buffer(pos, end - pos), yield[ec]);
				pos += n;
				if (ec) {
					BOOST_LOG_TRIVIAL(warning) <<"Error on receive data from aserver";
					break;
				}

				auto itr = std::search(first, pos, K_startid_str.begin(), K_startid_str.end());
				if (itr == pos) continue; // for more data

				itr += K_startid_str.size();
				// check and read header
				if (size_t(pos - itr) < K_header_size){
					n = boost::asio::async_read(*socket, boost::asio::buffer(pos, K_header_size - (pos - itr)), yield[ec]);
					pos += n;
					if (ec) {
						BOOST_LOG_TRIVIAL(warning) <<"Error on receive command header from aserver";
						break;
					}
				}
				// check and read data
				auto head = *reinterpret_cast<Header_t*>(itr);

				auto packageSize =  K_header_size + head.length + 1;
				if (size_t(pos - itr) < packageSize){
					n = boost::asio::async_read(*socket, boost::asio::buffer(pos, packageSize - (pos - itr)), yield[ec]);
					pos += n;
					if (ec) {
						BOOST_LOG_TRIVIAL(warning) <<"Error on receive command data  from aserver";
						break;
					}
				}

				unsigned char end_flag(*(itr + packageSize - 1));
				if(end_flag != 0xf0){
					BOOST_LOG_TRIVIAL(warning) <<"Bad end flag: 0x" << std::hex << uint32_t(end_flag);
					break;
				}

				std::string text(itr + K_header_size, itr + packageSize -1);
				auto payloadSize = itr - first + packageSize;
				std::copy(first + payloadSize, pos, v.begin());
				pos -= payloadSize;
				itr = first;

				if (head.command == 0xe0){
					if (text != "ASHeart"){
						BOOST_LOG_TRIVIAL(warning) <<"Bad heartbeat: " << text << text.size();
						break;
					}
				}else if (head.command == 0xe1){
					app_t app;
					std::string which_app;
					if (!parse_json(app, text, which_app)){
						BOOST_LOG_TRIVIAL(warning) <<"Failed to parse JSON: " << text;
						break;
					}
					if (which_app != "immeAPP")	continue;

					BOOST_LOG_TRIVIAL(info) <<"OK JSON: " << text;
					context.ios_aserver.post([=]{context.database->saveDeviceRecord(app);});
				}
				else {
					BOOST_LOG_TRIVIAL(warning) <<"Ignore bad command: 0x" << std::hex << unsigned(head.command);
				}
			}
		}
		if (socket && socket->is_open()){
			socket->close();
		}
		sleep_for_(K_retry_sleep_time, yield);
	}
	void LORAUpstream::run_send_(boost::asio::yield_context yield, std::shared_ptr<tcp::socket> socket){
		boost::system::error_code ec;
		boost::asio::steady_timer hb_deadline(context.ios_aserver);
		auto lora_heartbeat = std::chrono::seconds(context.config.lora_heartbeat);
		auto next_heartbeat = std::chrono::steady_clock::now();
		while(socket->is_open()){
			if (next_heartbeat <= std::chrono::steady_clock::now()){
				boost::asio::async_write(*socket, boost::asio::buffer(K_CSHeart), yield[ec]);	// heart beat
				if (ec){
					BOOST_LOG_TRIVIAL(warning) <<"Failed to send heartbeat." << ec.message();
					socket->close();
					return;
				}
				next_heartbeat += lora_heartbeat;
			}
			else if (mData.empty()){
				hb_deadline.expires_from_now(lora_heartbeat);
				hb_deadline.async_wait(yield[ec]);
			}
			else{
				boost::asio::async_write(*socket, boost::asio::buffer(*mData.begin()), yield[ec]);	// send down command
				if (ec){
					BOOST_LOG_TRIVIAL(warning) <<"Failed to send upstream data." << ec.message();
					socket->close();
					return;
				}
				mData.pop_front();
			}
		}
	}
	void LORAUpstream::sleep_for_(int sec, boost::asio::yield_context yield){
		boost::asio::steady_timer timer(context.ios_aserver);
		timer.expires_from_now(std::chrono::seconds(sec));
		boost::system::error_code ec;
		timer.async_wait(yield[ec]);
	}
	void LORAUpstream::append(std::vector<char>&& data){
		context.ios_aserver.dispatch([&]{
				mData.emplace_back(std::move(data));
				hb_deadline.cancel();
				});

	}
}

